package com.ford.cn.csre.scenerio.domain.vo;



import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.io.*;
import java.util.Date;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicInteger;

public class Demo02 {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 非常关键，一定要设置启动检查点！！
        env.enableCheckpointing(5000);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE); //精确一次

        env.setParallelism(1);
        //告警流
        DataStream<Alarm> dataStream = env.addSource(new MyAlarmSource());
        //规则流
        DataStream<Rule> ruleStream = env.addSource(new MyRuleSource());

        //dataStream.print();
        //ruleStream.print();
        dataStream.connect(ruleStream).flatMap(new RuleAlarmFunction()).print();

        env.execute("Drools Demo01");
    }
    public static void getKafKaStreamSource(){
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "127.0.0.1:9092");
        properties.setProperty("group.id", "g2");    // 第 1 个参数是固定值 group.id，第 2 个参数是自定义的组 ID
        DeserializationSchema<String> deserializationSchema = new SimpleStringSchema();
        String topic = "topic2";
        //new FlinkKafkaConsumer011<String>(topic, deserializationSchema, properties);

    }
    //自定义类，继承SourceFunction，每条数据是一个规则对象（根据业务自定义）
    public static class MyAlarmSource implements SourceFunction<Alarm>{
        // 定义一个标识，表示数据源是否继续运行
        private Boolean running = true;
        private static AtomicInteger counter = new AtomicInteger(0);


        @Override
        public void run(SourceContext<Alarm> sourceContext) throws Exception {
            String[] ids = {"111","222","333"};
            while (running){
                Alarm alarm = new Alarm();
                alarm.setAlarmId(ids[counter.getAndIncrement()%3]);
                alarm.setEventTime(new Date());
                Thread.sleep(500);
                sourceContext.collect(alarm);
            }
        }

        @Override
        public void cancel() {
            running = false;
        }
    }

    //自定义类，继承SourceFunction，每条数据是一个规则对象（根据业务自定义）
    public static class MyRuleSource implements SourceFunction<Rule>{
        // 定义一个标识，表示数据源是否继续运行
        private boolean running = true;

        @Override
        public void run(SourceContext<Rule> sourceContext) throws Exception {
            try (InputStream resource = this.getClass().getResourceAsStream("/rules2.drl")) {
                InputStreamReader read = new InputStreamReader(resource, "utf-8");
                BufferedReader bufferedReader = new BufferedReader(read);
                StringBuilder sb = new StringBuilder(1024);
                String lineTxt = null;
                while ((lineTxt = bufferedReader.readLine()) != null) {
                    sb.append(lineTxt).append("\r\n");
                }
                while (running) {
                    //counter = 1;
                    Thread.sleep(500);
                    Rule rule = new Rule();
                    rule.setDrlStr(sb.toString());
                    rule.setOperate(1);
                    rule.setName("rule2");
                    sourceContext.collect(rule);
                }
            }
        }

        @Override
        public void cancel() {
            this.running = false;
        }
    }

}